package com.rizzo.analytics.worker.redis.task;

import com.rizzo.infrastructure.redis.key.RedisKey;
import org.joda.time.DateTime;
import org.joda.time.Period;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;

public abstract class RedisTask {

    private static final Logger LOGGER = LoggerFactory.getLogger(RedisTask.class);

    protected RedisTemplate<RedisKey, String> redisKeyTemplate;

    private Period lockExpiration;

    protected RedisTask(RedisTemplate<RedisKey, String> redisKeyTemplate, Period lockExpiration) {
        this.lockExpiration = lockExpiration;
        this.redisKeyTemplate = redisKeyTemplate;
    }

    public abstract void perform();

    protected Boolean acquireLock(final RedisKey redisKey) {
        LOGGER.debug("Acquiring lock for redisKey: " + redisKey);
        // get the time of the lock to see if it has expired
        String instant = redisKeyTemplate.opsForValue().get(redisKey);
        if (instant != null) {
            // has the lock expired?
            DateTime expiration = new DateTime(Long.parseLong(instant));
            DateTime now = new DateTime();
            if (expiration.isBefore(now)) {
                LOGGER.debug("Lock for redisKey has expired trying to get the lock before anybody else does lock the lock: " + redisKey);
                // this means we can getset the lock with a new expiration and continue processing if the old value is still expired from the getset
                DateTime expirationTime = now.plus(lockExpiration);
                String oldExpiration = redisKeyTemplate.opsForValue().getAndSet(redisKey, Long.toString(expirationTime.getMillis()));
                DateTime oldExpirationTime = new DateTime(Long.parseLong(oldExpiration));
                // is the old expiration from the getset still expired -> if so we have the lock
                return oldExpirationTime.isBefore(now);
            } else {
                // lock is not expired so not locking...
                LOGGER.debug("Lock is not expired so keeping hands of: " + redisKey);
                return false;
            }
        } else {
            // no lock was present so setting it...
            LOGGER.debug("No lock present so setting the lock: " + redisKey);
            return redisKeyTemplate.execute(new RedisCallback<Boolean>() {
                @Override
                public Boolean doInRedis(RedisConnection connection) throws DataAccessException {
                    DateTime now = new DateTime();
                    DateTime expiration = now.plus(lockExpiration);
                    byte[] key = redisKey.serialize().getBytes();
                    byte[] expire = Long.toString(expiration.getMillis()).getBytes();
                    return connection.setNX(key, expire);
                }
            });
        }
    }

    protected void unlock(RedisKey redisKey) {
        String instant = redisKeyTemplate.opsForValue().get(redisKey);
        if (instant != null) {
            DateTime expiration = new DateTime(Long.parseLong(instant));
            DateTime now = new DateTime();
            if (expiration.isBefore(now)) {
                // this means the calculation took longer than the lockExpiration time... not good!
                LOGGER.warn(
                        "Unlocking of key happened after expiration: for the comfortable working of the trend-calculation increase the lockExpiration time!");
            } else {
                // unlock the key by deleting it so that others can take it
                LOGGER.debug("All work done now deleting the lock: " + redisKey);
                redisKeyTemplate.delete(redisKey);
            }
        } else {
            LOGGER.warn("Very weird: we can't find the lock that we initially created!");
        }
    }

}
